package com.rem.flink.flink3Transform;


import com.rem.flink.flink2Source.ClickSource;
import com.rem.flink.flink2Source.Event;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 数据转换 filter map flatmap操作
 *
 * @author Rem
 * @date 2022/10/9
 */
public class TransformMapTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //env.setParallelism(1);

        //有了自定义的source function，调用addSource方法
        DataStreamSource<Event> stream = env.addSource(new ClickSource());

        //filter操作
        //filter(stream);

        //map操作
        //map(stream);

        //flatmap操作
        //flatMap(stream);

        env.execute();
    }

    private static void flatMap(DataStreamSource<Event> stream) {
        stream.flatMap((event, collector) -> {
                    if ("Bob".equals(event.getUser())) {
                        collector.collect(event.getUser() + ":" + event.getUrl());
                    }
                })
                .returns(new TypeHint<>() {
                })
                .print();
    }

    private static void filter(DataStreamSource<Event> stream) {
        stream.filter(a -> "Bob".equals(a.getUser())).print();
    }

    private static void map(DataStreamSource<Event> stream) {
        stream.map(Event::getUser).print();
    }

}

